package com.sl.core.engine.meta.trans;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.ReflectUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sl.core.engine.ProcessConfig;
import com.sl.core.engine.domain.ProcessStepPauseConfig;
import com.sl.core.engine.meta.row.ProcessRowDescriptor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;
import org.apache.hop.core.IProgressMonitor;
import org.apache.hop.core.exception.HopMissingPluginsException;
import org.apache.hop.core.exception.HopTransformException;
import org.apache.hop.core.exception.HopXmlException;
import org.apache.hop.core.logging.LogChannel;
import org.apache.hop.core.row.IRowMeta;
import org.apache.hop.core.row.IValueMeta;
import org.apache.hop.core.row.RowMeta;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.core.variables.Variables;
import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider;
import org.apache.hop.pipeline.Pipeline;
import org.apache.hop.pipeline.PipelineMeta;
import org.apache.hop.pipeline.transform.TransformErrorMeta;
import org.apache.hop.pipeline.transform.TransformMeta;

import java.io.InputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


@NoArgsConstructor
public class ProcessTransDescriptor extends PipelineMeta {
    private static final Class<?> PKG = Pipeline.class; // For Translator

    /**
     * 流程版本
     */
    @Getter
    @Setter
    private String processName;
    @Getter
    @Setter
    private String processVersion;


    @Getter
    @Setter
    private String description;
    @Getter
    @Setter
    private String processCode;

    @Getter
    @Setter
    private String processId;

    @Getter
    @Setter
    private String openId;

    @Getter
    @Setter
    private Boolean rowLogFlag = true;

    private static final Map<String, List<ProcessStepPauseConfig>> stepPauseMap = Maps.newHashMap();

    public ProcessStepPauseConfig getStepPause(String tid, String stepId) {
        List<ProcessStepPauseConfig> processStepPauseConfigs = stepPauseMap.get(tid);
        if (CollUtil.isEmpty(processStepPauseConfigs)) {
            return null;
        }
        return processStepPauseConfigs.stream().filter(v -> StrUtil.equals(v.getStepId(), stepId)).findFirst().orElse(null);
    }

    public ProcessTransDescriptor(InputStream xmlStream) throws HopXmlException, HopMissingPluginsException {
        this(xmlStream, new HashMap<>(), null);
    }

    public ProcessTransDescriptor(InputStream xmlStream, Map<String, Object> variableMap) throws HopXmlException, HopMissingPluginsException {
        this(xmlStream, variableMap, null);
    }

    public ProcessTransDescriptor(InputStream xmlStream, Map<String, Object> variableMap, ProcessConfig processConfig) throws HopXmlException, HopMissingPluginsException {
        super(xmlStream, new MemoryMetadataProvider(), new Variables());


        if (ObjectUtil.isNull(processConfig)) {
            this.setName("default");
            this.setOpenId("default");
            return;
        }
        this.setName(processConfig.getProcessName());
        this.setProcessName(processConfig.getProcessName());
        this.setProcessCode(processConfig.getProcessCode());
        this.setProcessVersion(processConfig.getProcessVersion());
        this.setDescription(processConfig.getDescription());
        this.setOpenId(processConfig.getOpenId());
    }


    public static void addStepPause(String tid, String stepId) {
        addStepPause(tid, stepId, null);
    }

    public static void addStepPause(String tid, String stepId, String condition) {
        List<ProcessStepPauseConfig> strings = stepPauseMap.computeIfAbsent(tid, (v) -> {
            return Lists.newArrayList();
        });

        ProcessStepPauseConfig processStepPauseConfig = new ProcessStepPauseConfig();
        processStepPauseConfig.setStepId(stepId);
        processStepPauseConfig.setCondition(condition);
        strings.add(processStepPauseConfig);
    }

    public void stepResume() {
        stepPauseMap.clear();
    }

    public void stepResume(String tid, String stepId) {
        List<ProcessStepPauseConfig> strings = stepPauseMap.computeIfAbsent(tid, (v) -> {
            return Lists.newArrayList();
        });

        strings.removeIf(v -> StrUtil.equals(stepId, v.getStepId()));
    }

    public void stepResume(String tid) {
        stepPauseMap.remove(tid);

    }

    @Override
    public IRowMeta getTransformFields(
            IVariables variables,
            TransformMeta transformMeta,
            TransformMeta targetTransform,
            IProgressMonitor monitor)
            throws HopTransformException {
        IRowMeta row = new ProcessRowDescriptor();

        if (transformMeta == null) {
            return row;
        }

        String fromToCacheEntry =
                transformMeta.getName()
                        + (targetTransform != null ? ("-" + targetTransform.getName()) : "");
        IRowMeta rowMeta = transformFieldsCache.get(fromToCacheEntry);
        if (rowMeta != null) {
            return rowMeta;
        }

        // See if the transform is sending ERROR rows to the specified target transform.
        //
        if (targetTransform != null && transformMeta.isSendingErrorRowsToTransform(targetTransform)) {
            // The error rows are the same as the input rows for
            // the transform but with the selected error fields added
            //
            row = getPrevTransformFields(variables, transformMeta);

            // Check if row object is null that means the transform could be an input transform. In this
            // case, get fields
            // out from transform itself
            if (row.isEmpty()) {
                row = getThisTransformFields(variables, transformMeta, targetTransform, row, monitor);
            }

            // Add to this the error fields...
            TransformErrorMeta transformErrorMeta = transformMeta.getTransformErrorMeta();
            row.addRowMeta(transformErrorMeta.getErrorRowMeta(variables));

            // Store this row in the cache
            //
            transformFieldsCache.put(fromToCacheEntry, row);

            return row;
        }

        // Resume the regular program...

        List<TransformMeta> prevTransforms = findPreviousTransforms(transformMeta, false);

        int nrPrevious = prevTransforms.size();

        if (LogChannel.GENERAL.isDebug()) {
            LogChannel.GENERAL.logDebug(
                    BaseMessages.getString(
                            PKG,
                            "PipelineMeta.Log.FromTransformALookingAtPreviousTransform",
                            transformMeta.getName(),
                            String.valueOf(nrPrevious)));
        }
        for (int i = 0; i < prevTransforms.size(); i++) {
            TransformMeta prevTransformMeta = prevTransforms.get(i);

            if (monitor != null) {
                monitor.subTask(
                        BaseMessages.getString(
                                PKG,
                                "PipelineMeta.Monitor.CheckingTransformTask.Title",
                                prevTransformMeta.getName()));
            }

            IRowMeta add = getTransformFields(variables, prevTransformMeta, transformMeta, monitor);
            if (add == null) {
                add = new RowMeta();
            }
            if (LogChannel.GENERAL.isDebug()) {
                LogChannel.GENERAL.logDebug(
                        BaseMessages.getString(PKG, "PipelineMeta.Log.FoundFieldsToAdd") + add.toString());
            }
            if (i == 0) {
                row.addRowMeta(add);
            } else {
                // See if the add fields are not already in the row
                for (int x = 0; x < add.size(); x++) {
                    IValueMeta v = add.getValueMeta(x);
                    IValueMeta s = row.searchValueMeta(v.getName());
                    if (s == null) {
                        row.addValueMeta(v);
                    }
                }
            }
        }

        // Finally, see if we need to add/modify/delete fields with this transform "name"
        rowMeta = getThisTransformFields(variables, transformMeta, targetTransform, row, monitor);

        // Store this row in the cache
        //
        transformFieldsCache.put(fromToCacheEntry, rowMeta);

        return rowMeta;
    }


    @Override
    public IRowMeta getPrevTransformFields(
            IVariables variables,
            TransformMeta transformMeta,
            final String transformName,
            IProgressMonitor monitor)
            throws HopTransformException {
        ReflectUtil.invoke(this, "clearTransformFieldsCache");
        IRowMeta row = new ProcessRowDescriptor();

        if (transformMeta == null) {
            return null;
        }
        List<TransformMeta> prevTransforms = findPreviousTransforms(transformMeta);
        int nrPrevTransforms = prevTransforms.size();
        if (LogChannel.GENERAL.isDebug()) {
            LogChannel.GENERAL.logDebug(
                    BaseMessages.getString(
                            PKG,
                            "PipelineMeta.Log.FromTransformALookingAtPreviousTransform",
                            transformMeta.getName(),
                            String.valueOf(nrPrevTransforms)));
        }
        TransformMeta prevTransformMeta;
        for (int i = 0; i < nrPrevTransforms; i++) {
            prevTransformMeta = prevTransforms.get(i);
            if (transformName != null && !transformName.equalsIgnoreCase(prevTransformMeta.getName())) {
                continue;
            }

            if (monitor != null) {
                monitor.subTask(
                        BaseMessages.getString(
                                PKG,
                                "PipelineMeta.Monitor.CheckingTransformTask.Title",
                                prevTransformMeta.getName()));
            }

            IRowMeta add = getTransformFields(variables, prevTransformMeta, transformMeta, monitor);

            if (LogChannel.GENERAL.isDebug()) {
                LogChannel.GENERAL.logDebug(
                        BaseMessages.getString(PKG, "PipelineMeta.Log.FoundFieldsToAdd2") + add.toString());
            }
            if (i == 0) {
                // we expect all input streams to be of the same layout!

                row.addRowMeta(add); // recursive!
            } else {
                // See if the add fields are not already in the row
                for (int x = 0; x < add.size(); x++) {
                    IValueMeta v = add.getValueMeta(x);
                    IValueMeta s = row.searchValueMeta(v.getName());
                    if (s == null) {
                        row.addValueMeta(v);
                    }
                }
                if(row instanceof ProcessRowDescriptor processRowDescriptor){
                    processRowDescriptor.copyField(add);
                }

            }
        }
        return row;
    }

}
